package com.atguigu.gmall.realtime.app.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.DimJoinFunction;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @author Felix
 * @date 2023/4/25
 * 发送异步请求进行维度关联
 * 模板方法设计模式：
 *      在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是有些步骤在父类中没办法实现，可以延迟到子类中去实现
 *      在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    private ExecutorService executorService;

    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
    }

    //发送异步请求，完成维度关联
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //开启线程
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                //根据流中的对象获取要关联的维度主键
                String key = getKey(obj);
                //根据维度的主键获取维度对象
                JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                //将维度对象的属性补充到流中对象上
                if(dimInfoJsonObj != null){
                    join(obj,dimInfoJsonObj);
                }
                //将异步的响应结果交给resultFuture的回调函数进行处理--向下游传递数据
                resultFuture.complete(Collections.singleton(obj));
            }
        });
    }
}
